package cn.itcast.up.model.matchtag

import java.util.Properties

import cn.itcast.up.model.bean.{HBaseMeta, TagRule}
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 职业标签主程序
  */
object JobModel {

  def main(args: Array[String]): Unit = {
    //    1. 构建一个SparkSession的对象
    val spark: SparkSession = SparkSession.builder()
      .appName("JobModel")
      .master("local[*]")
      .getOrCreate()
    //导入隐式类型转换
    import spark.implicits._
    import org.apache.spark.sql.functions._
    //2. 加载MySQL数据源,整个数据库中的表数据都可以获取到了.
    val url = "jdbc:mysql://bd001:3306/tags_new?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    val tableName = "tbl_basic_tag"
    val props = new Properties
    val mysqlSource: DataFrame = spark.read.jdbc(url, tableName, props)

    //3. 获取tbl_basic_tag中的数据:
    //   1. 4级标签的数据:id,rule:数据源信息,将数据封装为HBaseMeta对象
    val meta: HBaseMeta = mysqlSource.select('id, 'rule)
      .where("id = 476")
      .map(row => {
        //    inType=HBase##zkHosts=192.168.10.20##zkPort=2181##hbaseTable=tbl_users##family=detail##selectFields=id,job
        val ruleStr: String = row.getAs[String]("rule")
        val tuples: Array[(String, String)] = ruleStr.split("##")
          .map(kv => {
            val arr: Array[String] = kv.split("=")
            (arr(0), arr(1))
          })
        val ruleMap: Map[String, String] = tuples.toMap
        HBaseMeta(ruleMap)
      }).collect()(0)
//    println(meta)
//    HBaseMeta(HBase,192.168.10.20,2181,tbl_users,detail,id,job)

    //   2. 5级标签的数据:id,rule: 匹配的时候使用.
    val fiveRule: List[TagRule] = mysqlSource.select('id, 'rule)
      .where("pid = 476")
      .map(row => {
        val id: String = row.getAs[Long]("id").toString
        val rule: String = row.getAs[String]("rule").toString
        TagRule(id, rule)
      }).collect().toList
//    println(fiveRule)

    //4. 获取HBase中的数据
    //   1. 使用自定义数据源加载HBase中的数据.
    val hbaseSource: DataFrame = spark.read
      //自定义数据源的全类名.
      .format("cn.itcast.up.model.tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, meta.zkHosts)
      .option(HBaseMeta.ZKPORT, meta.zkPort)
      .option(HBaseMeta.HBASETABLE, meta.hbaseTable)
      .option(HBaseMeta.FAMILY, meta.family)
      .option(HBaseMeta.SELECTFIELDS, meta.selectFields)
      .load()
//    hbaseSource.show()
//    +---+---+
    //| id|job|
    //+---+---+
    //|  1|  3|
    //| 10|  5|
    //|100|  3|
    //|101|  1|

    //5. 使用5级规则数据和HBase中的源数据进行匹配,找到用户所属的标签.

    val getTagId = udf((job: String) => {
      //定义标签id
      var tagId = ""
      //遍历集合
      for (tagRule <- fiveRule) {
      //找到对应的标签
        if (tagRule.rule.equals(job)) {
          tagId = tagRule.id
        }
      }
      tagId
    })

    val result: DataFrame = hbaseSource.select('id.as("userid"), getTagId('job).as("tagIds"))
//    result.show()
//+------+------+
    //|userid|tagIds|
    //+------+------+
    //|     1|   479|
    //|    10|   481|
    //|   100|   479|
    //|   101|   477|

    //6. 将结果存入HBase.
    //   1. 使用自定义数据源将表情结果存入HBase
    result.write
      .format("cn.itcast.up.model.tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, meta.zkHosts)
      .option(HBaseMeta.ZKPORT, meta.zkPort)
      .option(HBaseMeta.HBASETABLE, "test32")
      .option(HBaseMeta.FAMILY, meta.family)
//      .option(HBaseMeta.SELECTFIELDS, "userid,tagIds")
      .save()
  }
}
